/**
 * 
 */
package com.ms.service.module.gamelog.async;

import java.math.BigDecimal;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.hibernate.HibernateException;
import org.hibernate.Session;
import org.hibernate.Transaction;

import redis.clients.jedis.Jedis;

import com.ms.game.log.GameWriteLogAsyncTask;
import com.ms.service.db.jedis.JedisManager;
import com.ms.service.module.gamelog.GameLogModule;
import com.xpec.c4.db.hibernate.manager.HibernateLogManager;
import com.xpec.c4.game.common.Helper;
import com.xpec.c4.game.utility.GameUtility;
import com.xpec.c4.game.utility.JsonConvert;
import com.xpec.c4.service.module.task.TaskAsyncCommand;

/**
 * 處理將暫存 log 檔內容寫回 dv 的異步處理 task
 * 
 * @author TW15010
 */
public class LogProcTaskAsync2 extends TaskAsyncCommand {
	private static final Log log = LogFactory.getLog(LogProcTaskAsync2.class);

	/**
	 * 每次從 redis 中讀取的 log 數量
	 */
	public static final int READ_DATA_COUNT = 200;

	/**
	 * 累積多少筆資料 commit 一次
	 */
	public static final int DATA_COUNT_PER_COMMIT = 100;

	/**
	 * 在 log module 佔用第幾個處理位置, 跟 queue id 一樣，但是給 log module 那一層看的
	 */
	private int procId;

	/**
	 * 這個 module 使用的 log redis 資源號碼
	 */
	private int logRedisNum = 0;

	/**
	 * 執行這個 task 的 log module 實體
	 */
	private GameLogModule gameLogModule;

	/**
	 * @param procId
	 * @param file
	 * @param gameLogModule
	 */
	public LogProcTaskAsync2(int procId, int logRedisNum, GameLogModule gameLogModule) {
		super();
		this.procId = procId;
		this.logRedisNum = logRedisNum;
		this.gameLogModule = gameLogModule;
	}

	@Override
	public void asyncExecute() throws Exception {
		// 取得 redis 資源
		int readCnt = 0;
		Jedis jedis = JedisManager.getLogResource(logRedisNum);

		// 還原寫回 db ======================================
		long procBegTime = System.currentTimeMillis();
		int logCnt = 0;
		int saveCnt = 0;

		Session session = HibernateLogManager.getLogSession();
		Transaction tx = session.getTransaction();

		procBegTime = System.currentTimeMillis();

		boolean isContinue = true;
		String line;
		while (isContinue) {
			// 本來是使用一口氣先讀一定數量出來再一次寫入 db, 但這樣會有某幾個 game log module 總是一次讀到滿,
			// 壓力會特別大, 最後改成讀一筆寫一筆的方式處理 log string
			try {
				line = jedis.lpop(GameWriteLogAsyncTask.REDIS_LOG_LIST_KEY);

				// 取不到時就結束
				if (!GameUtility.isHaveString(line)) {
					isContinue = false;
					continue;
				}

				// 增加讀取數量
				readCnt++;

				// 還原成 model
				Object logObject = JsonConvert.toObject(line);

				// 成功將 log model 還原
				if (logObject != null) {
					if (saveCnt == 0) {
						tx.begin();
					}

					// 存入 log db
					try {
						session.save(logObject);
						session.flush();

						// 暫停一下讓出 cpu 資源
						Thread.sleep(1);
					} catch (Exception e1) {
						if (log.isErrorEnabled()) {
							log.error("save/flush " + logObject.getClass().getSimpleName() + " fail. ", e1);
							log.error("log detail = " + line);
							continue;
						}
					}
					saveCnt++;
					logCnt++;

					// 寫入一定數量後 commit 一次
					if (saveCnt >= DATA_COUNT_PER_COMMIT) {
						try {
							tx.commit();
							session.clear();
						} catch (HibernateException e) {
							tx.rollback();
							if (log.isErrorEnabled()) {
								log.error("commit fail.", e);
							}
						}
						saveCnt = 0;
					}
				} else { // if (logObject != null) else
					if (log.isErrorEnabled()) {
						log.error("convert to model fail");
						log.error("log detail = " + line);
					}
				}
			} catch (Exception e) {
				if (log.isErrorEnabled()) {
					log.error("logFilesProc() fail.", e);
				}
			}

			// 處理到指定的數量就結束
			if (readCnt >= READ_DATA_COUNT) {
				isContinue = false;
				continue;
			}
		} // while ((line = br.readLine()) != null) {

		// 結束後有剩下的再 commit 一次
		if (saveCnt > 0) {
			try {
				tx.commit();
				session.clear();
			} catch (HibernateException e) {
				tx.rollback();
				if (log.isErrorEnabled()) {
					log.error("commit fail.", e);
				}
			}
		}

		if (readCnt > 0) {
			if (log.isInfoEnabled()) {
				long timeDiff = System.currentTimeMillis() - procBegTime;
				BigDecimal speed = Helper.div(timeDiff, logCnt);
				BigDecimal procPerSec = Helper.div(new BigDecimal(1000), speed);
				log.info("read from redis " + logRedisNum + ", read " + readCnt + " import " + logCnt
						+ " logs to DB " + timeDiff + " ms " + procPerSec + "/sec");
			}
		}

		// 釋放資源
		JedisManager.returnLogResource(logRedisNum, jedis);
		HibernateLogManager.endSession();
	}

	@Override
	public void execute() throws Exception {
		if (gameLogModule != null) {
			gameLogModule.onLogProcTaskAsyncFinish(this);
		}
	}

	public int getProcId() {
		return procId;
	}

	public GameLogModule getGameLogModule() {
		return gameLogModule;
	}
}
